package com.yunhe.transfer.mqtt;

import cn.beepower.cloud.msg.CloudMeasureValue;
import cn.beepower.cloud.msg.CloudMeasureValues;
import cn.beepower.domain.bus.MqttDataBus;
import cn.beepower.domain.bus.ThreadMode;
import cn.beepower.domain.message.BeeMessage;
import cn.beepower.domain.message.Payload;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.model.authority.Station;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.transfer.client.MeasHistoryClient;
import com.yunhe.common.constant.CacheConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.constant.MqttConsts;
import com.yunhe.common.model.DomainList;
import com.yunhe.common.model.authority.StationDTO;
import com.yunhe.common.model.base.DataTransferDTO;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.transfer.proto.EigProto;
import com.yunhe.transfer.service.base.DataTransferService;
import com.yunhe.common.util.*;
import com.wanke.scada.mqtt.MessageKit;
import com.yunhe.transfer.util.MeasurementUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;

/**
 * @title: MeasurementMqttHandler
 * @Author Chen Lun
 * @Date: 2021/1/21
 */
@Component
public class MeasurementMqttHandler {

    @Autowired
    @Qualifier("MqttDataBus")
    private MqttDataBus dataBus;
    @Autowired
    private MqttClient mqttClient;
    @Autowired
    private DataTransferService dataTransferService;
    @Autowired
    private MeasHistoryClient measHistoryClient;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private RedisClient redisClient;
    @Autowired
    private AuthorityClient authorityClient;

    private Logger logger = LoggerFactory.getLogger(MeasurementMqttHandler.class);

    private static final ObjectMapper MAPPER = JsonUtil.getObjectMapper();

    /**
     * MQTT主动补招消息发送
     * @param record
     * @throws IOException
     */
    @KafkaListener(topics = KafkaConsts.Topic.RecollectOrder, groupId = "mqttRecollectOrder")
    public void publishRecollect(ConsumerRecord<?, String> record) throws IOException {
        String value = record.value();
        if (!StringUtil.isEmpty(value)) {
            DataTransferDTO dataMap = MAPPER.readValue(value, DataTransferDTO.class);
            logger.info("接收到主动补招消息：{}", dataMap.toString());
            Long stationCode = dataMap.getStationCode();
            String topic = MqttConsts.TOPIC.OrderCollect.value(stationCode);
            synchronized (topic) {
                dataBus.publish(topic, MAPPER.writeValueAsString(dataMap).getBytes(), 1);
            }
        }
    }

    /**
     * 数据传输主题订阅
     * @param stationCode
     */
    public void subscribeCollect(Long stationCode) {
        dataBus.subscribe(MqttConsts.TOPIC.Collect.value(stationCode), 1, ThreadMode.Async, msg -> {
            BeeMessage message = msg.getM();
            String topic = message.getTopic();
            logger.info("订阅数据接收：{}", topic);
            try {
                processMessage(message, 1, stationCode);
            } catch (Exception e) {
                logger.error("Mqtt接收消息出错，主题：{}", topic, e);
            }
            return null;
        });
    }

    /**
     * 主动补招数据主题订阅
     * @param stationCode
     */
    public void subscribeOrderCollect(Long stationCode) {
        dataBus.subscribe(MqttConsts.TOPIC.Recollect.value(stationCode), 1, ThreadMode.Async, msg -> {
            BeeMessage message = msg.getM();
            String topic = message.getTopic();
            logger.info("订阅数据补招：{}", topic);
            try {
                processMessage(message, 2, stationCode);
            } catch (Exception e) {
                logger.error("数据补招报错：{}", e);
            }
            return null;
        });
    }

    /**
     * 订阅DPU量测数据上传上传
     * @param beeId
     */
    public void subscribeDpuData(Long stationCode, String timeZone, String beeId) {
        try {
            mqttClient.subscribe(MqttConsts.TOPIC.DPU_SM.value(beeId), 1, (topic, msg) -> processDpuMsg(stationCode, timeZone, msg));
            mqttClient.subscribe(MqttConsts.TOPIC.DPU_AM.value(beeId), 1, (topic, msg) -> processDpuMsg(stationCode, timeZone, msg));
        } catch (Exception e) {
            logger.error("订阅DPU数据上传出错", e);
        }
    }

    /**
     * 处理DPU量测数据上传
     */
    private void processDpuMsg(Long stationCode, String timeZone, MqttMessage msg) {
        try {
            EigProto.pbPointValues pbPointValues = EigProto.pbPointValues.parseFrom(msg.getPayload());
            List<EigProto.pbAnalogValue> valuesList = pbPointValues.getAValuesList();
            if (valuesList != null) {
                List<Measurement> measurements = new ArrayList<>();
                for (EigProto.pbAnalogValue pbAnalogValue : valuesList) {
                    Measurement measurement = new Measurement();
                    measurement.setPointNumber(Long.parseLong(stationCode + "" + pbAnalogValue.getPointId()));
                    measurement.setRtime(pbAnalogValue.getTimestamp());
                    measurement.setDtime(DateTimeUtils.getDateString(pbAnalogValue.getTimestamp(), timeZone));
                    measurement.setVal(pbAnalogValue.getMeasValue());
                    measurements.add(measurement);
                }
                receiveData(measurements, stationCode);
            }
        } catch (Exception e) {
            logger.error("处理DPU量测数据上传出错，电站：{}", stationCode, e);
        }
    }

    /**
     * 处理订阅收到的信息
     * @param beeMessage
     * @param type
     * @param stationCode
     * @throws IOException
     */
    private void processMessage(BeeMessage beeMessage, Integer type, Long stationCode) throws IOException {
        Payload payload = beeMessage.getPayload();
        byte[] payloadByte = payload.getValue().toByteArray();
        CloudMeasureValues cloudMeasureValues = new CloudMeasureValues();
        MessageKit.mergeFrom(cloudMeasureValues, payloadByte);
        List<Measurement> measurements = covertMessages(cloudMeasureValues, stationCode);
        if (measurements != null && measurements.size() > 0) {
            if (type == 1) {
                receiveData(measurements, stationCode);
            } else if (type == 2) {
                recollectData(measurements, stationCode);
            }
        }
    }

    /**
     * 数据传输MQTT主题订阅实现
     * @param measurements
     * @param stationCode
     * @throws JsonProcessingException
     */
    private void receiveData(List<Measurement> measurements, Long stationCode) throws JsonProcessingException {
        dataTransferService.receiveData(measurements, stationCode);
    }

    /**
     * 主动补招数据主题订阅实现
     * @param measurements
     * @param stationCode
     * @throws JsonProcessingException
     */
    private void recollectData(List<Measurement> measurements, Long stationCode) throws JsonProcessingException {
        DataTransferDTO dataTransferDTO = MeasurementUtil.recollectData(measurements, stationCode);
        saveHistory(measurements);
        String key = CacheConsts.KEY.RECOLLECT.value(UUID.randomUUID());
        redisClient.setObject(key, dataTransferDTO, CacheConsts.getDefaultTime());
        kafkaTemplate.send(KafkaConsts.Topic.Recollect, key);
    }

    /**
     * 保存历史量测
     *
     * @param measurements
     */
    private void saveHistory(List<Measurement> measurements) {
        DomainList<Measurement> domainList = new DomainList<>(measurements);
        measHistoryClient.save(domainList, MeasurementConsts.FREQUENCY.Original.value());
    }

    /**
     * 转换proto量测列表
     *
     * @param measureValue
     * @param stationCode 电站代码
     * @return
     */
    private List<Measurement> covertMessages(CloudMeasureValues measureValue, Long stationCode) {
        Station station = authorityClient.getStationById(stationCode).getResults();
        if (station != null) {
            List<Measurement> result = new LinkedList<>();
            if (measureValue.getValuesList() != null) {
                for (CloudMeasureValue cloudMeasureValue : measureValue.getValuesList()) {
                    Measurement measurement = covertMessage(cloudMeasureValue, station.getTimeZone());
                    result.add(measurement);
                }
            }
            return result;
        } else {
            logger.error("电站代码{}找不到对应电站", measureValue.getProjectCode());
            return null;
        }
    }

    /**
     * 转换proto量测
     *
     * @param cloudMeasureValue proto量测
     * @param timeZone          时区
     * @return
     */
    private Measurement covertMessage(CloudMeasureValue cloudMeasureValue, String timeZone) {
        Measurement measurement = new Measurement();
        measurement.setPointNumber(cloudMeasureValue.getPointNumber());
        measurement.setRtime(cloudMeasureValue.getDTime());
        measurement.setVal(cloudMeasureValue.getValue());
        measurement.setFlag(cloudMeasureValue.getFlag());
        measurement.setDtime(DateTimeUtils.getDateString(cloudMeasureValue.getDTime(), timeZone));
        return measurement;
    }
}
